package com.example.wan1.flowable;

import android.os.Bundle;
import android.util.Log;
import android.view.View;

import com.example.wan1.R;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import androidx.appcompat.app.AppCompatActivity;
import butterknife.ButterKnife;
import butterknife.OnClick;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/***
 * 控制 观察者接收事件的速度 是通过subscription.request()来实现的
 * 控制 被观察者发送事件的速度 是通过emitter.requested()来实现的
 * 上面这两种情况分为同线程下进行和不同线程下进行
 * 2*2=4 四种情况
 * 再加上五种策略
 *
 * 注意：不管何种线程下进行，都必须实现subscription.request()才能接受到事件
 */
public class ThirdActivity extends AppCompatActivity {

    private static final String TAG = "ThirdActivity";

    Subscription subscription;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_third);
        ButterKnife.bind(this);

        Observable.just(1,2,3).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    @OnClick({R.id.button, R.id.button1, R.id.button2, R.id.button3, R.id.button4, R.id.button5, R.id.button6, R.id.button7, R.id.button8, R.id.button9, R.id.button10})
    public void onViewClicked(View view) {
        switch (view.getId()) {
            case R.id.button:
                errorStrategy();
                break;
            case R.id.button1:
                missingStrategy();
                break;
            case R.id.button2:
                bufferStrategy();
                break;
            case R.id.button3:
                dropStrategy();
                break;
            case R.id.button4:
                latestStrategy();
                break;
            case R.id.button5:
                subscription.request(48);
                break;
            case R.id.button6:
                makeProblem();
                break;
            case R.id.button7:
                asyncRecive();
                break;
            case R.id.button8:
                syncRecive();
                break;
            case R.id.button9:
                asyncSend();
                break;
            case R.id.button10:
                syncSend();
                break;
        }
    }

    /***
     * 同步：控制 被观察者发送事件 的速度
     */
    private void syncSend() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {

                // 调用emitter.requested()获取当前观察者需要接收的事件数量
                long n = emitter.requested();

                Log.d(TAG, "观察者可接收事件" + n);

                // 根据emitter.requested()的值，即当前观察者需要接收的事件数量来发送事件
                for (int i = 0; i < n; i++) {
                    Log.d(TAG, "发送了事件" + i);
                    emitter.onNext(i);
                    //FlowableEmitter.requested()的第二种特性：实时更新性
                    Log.e(TAG, "发送了事件" + i + "后, 还需要发送事件数量 = " + emitter.requested());
                }

                //FlowableEmitter.requested()的第二种特性：异常
                //当还需要发送的事件变成0的时候，在发送事件就会报错，出现MissingBackpressureException
                emitter.onNext(100);
            }
        }, BackpressureStrategy.ERROR)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");

                        // 设置观察者每次能接受10个事件
                        s.request(10);
                        //FlowableEmitter.requested()的第一种特性：叠加
                        //第二次接受20个事件，加上之前的10个，一共30个
                        s.request(20);

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "接收到了事件" + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });
    }

    /***
     * 异步：控制 被观察者发送事件 的速度
     */
    private void asyncSend() {
        // 被观察者：一共需要发送500个事件，但真正开始发送事件的前提 = FlowableEmitter.requested()返回值 ≠ 0
// 观察者：每次接收事件数量 = 48（点击按钮）

        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {

                Log.d(TAG, "观察者可接收事件数量 = " + emitter.requested());
                boolean flag; //设置标记位控制

                // 被观察者一共需要发送500个事件
                for (int i = 0; i < 500; i++) {
                    flag = false;

                    // 若requested() == 0则不发送
                    while (emitter.requested() == 0) {
                        if (!flag) {
                            Log.d(TAG, "不再发送");
                            flag = true;
                        }
                    }
                    // requested() ≠ 0 才发送
                    Log.d(TAG, "发送了事件" + i + "，观察者可接收事件数量 = " + emitter.requested());
                    emitter.onNext(i);


                }
            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
                .observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        subscription = s;
                        // 初始状态 = 不接收事件；通过点击按钮接收事件
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "接收到了事件" + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });
    }

    /***
     * 异步：控制 观察者接收事件 的速度
     * 特点：发送事件到缓存区中了
     */
    private void asyncRecive() {
        // 1. 创建被观察者Flowable
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                // 一共发送4个事件
                Log.d(TAG, "发送事件 1");
                emitter.onNext(1);
                Log.d(TAG, "发送事件 2");
                emitter.onNext(2);
                Log.d(TAG, "发送事件 3");
                emitter.onNext(3);
                Log.d(TAG, "发送事件 4");
                emitter.onNext(4);
                Log.d(TAG, "发送完成");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
                .observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        // 对比Observer传入的Disposable参数，Subscriber此处传入的参数 = Subscription
                        // 相同点：Subscription参数具备Disposable参数的作用，即Disposable.dispose()切断连接, 同样的调用Subscription.cancel()切断连接
                        // 不同点：Subscription增加了void request(long n)

                        s.request(3);
                        // 作用：决定观察者能够接收多少个事件
                        // 如设置了s.request(3)，这就说明观察者能够接收3个事件（多出的事件存放在缓存区）
                        // 官方默认推荐使用Long.MAX_VALUE，即s.request(Long.MAX_VALUE);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "接收到了事件" + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });
    }
    //2020-05-06 22:23:24.560 12281-12337/com.example.rxmvp D/ThirdActivity: 发送事件 1
    //2020-05-06 22:23:24.560 12281-12337/com.example.rxmvp D/ThirdActivity: 发送事件 2
    //2020-05-06 22:23:24.560 12281-12337/com.example.rxmvp D/ThirdActivity: 发送事件 3
    //2020-05-06 22:23:24.560 12281-12337/com.example.rxmvp D/ThirdActivity: 发送事件 4
    //2020-05-06 22:23:24.560 12281-12337/com.example.rxmvp D/ThirdActivity: 发送完成
    //2020-05-06 22:23:24.563 12281-12281/com.example.rxmvp D/ThirdActivity: 接收到了事件1
    //2020-05-06 22:23:24.563 12281-12281/com.example.rxmvp D/ThirdActivity: 接收到了事件2
    //2020-05-06 22:23:24.563 12281-12281/com.example.rxmvp D/ThirdActivity: 接收到了事件3

    /***
     * 同步：控制 观察者接收事件 的速度
     * 特点：发一个接受一个。发送第四个的时候，出现MissingBackpressureException的异常
     */
    private void syncRecive() {
        // 1. 创建被观察者Flowable
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                // 一共发送4个事件
                Log.d(TAG, "发送事件 1");
                emitter.onNext(1);
                Log.d(TAG, "发送事件 2");
                emitter.onNext(2);
                Log.d(TAG, "发送事件 3");
                emitter.onNext(3);
                Log.d(TAG, "发送事件 4");
                emitter.onNext(4);
                Log.d(TAG, "发送完成");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        // 对比Observer传入的Disposable参数，Subscriber此处传入的参数 = Subscription
                        // 相同点：Subscription参数具备Disposable参数的作用，即Disposable.dispose()切断连接, 同样的调用Subscription.cancel()切断连接
                        // 不同点：Subscription增加了void request(long n)

                        s.request(3);
                        // 作用：决定观察者能够接收多少个事件
                        // 如设置了s.request(3)，这就说明观察者能够接收3个事件（多出的事件存放在缓存区）
                        // 官方默认推荐使用Long.MAX_VALUE，即s.request(Long.MAX_VALUE);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "接收到了事件" + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });
    }
    //2020-05-06 22:23:25.715 12281-12281/com.example.rxmvp D/ThirdActivity: 发送事件 1
    //2020-05-06 22:23:25.715 12281-12281/com.example.rxmvp D/ThirdActivity: 接收到了事件1
    //2020-05-06 22:23:25.715 12281-12281/com.example.rxmvp D/ThirdActivity: 发送事件 2
    //2020-05-06 22:23:25.715 12281-12281/com.example.rxmvp D/ThirdActivity: 接收到了事件2
    //2020-05-06 22:23:25.715 12281-12281/com.example.rxmvp D/ThirdActivity: 发送事件 3
    //2020-05-06 22:23:25.715 12281-12281/com.example.rxmvp D/ThirdActivity: 接收到了事件3
    //2020-05-06 22:23:25.715 12281-12281/com.example.rxmvp D/ThirdActivity: 发送事件 4
    //2020-05-06 22:23:25.717 12281-12281/com.example.rxmvp W/ThirdActivity: onError:
    //    io.reactivex.exceptions.MissingBackpressureException: create: could not emit value due to lack of requests
    //2020-05-06 22:23:25.717 12281-12281/com.example.rxmvp D/ThirdActivity: 发送完成

    /***
     * 出现问题
     */
    private void makeProblem() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            // 1. 创建被观察者 & 生产事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                for (int i = 0; ; i++) {
                    Log.d(TAG, "发送了事件" + i);
                    Thread.sleep(10);
                    // 发送事件速度：10ms / 个
                    emitter.onNext(i);

                }

            }
        }).subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
                .observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
                .subscribe(new Observer<Integer>() {
                    // 2. 通过通过订阅（subscribe）连接观察者和被观察者

                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Integer value) {

                        try {
                            // 接收事件速度：5s / 个
                            Thread.sleep(5000);
                            Log.d(TAG, "接收到了事件" + value);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "对Error事件作出响应");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "对Complete事件作出响应");
                    }

                });
    }

    /***
     * 不使用背压策略是不行的，Mode不能为空
     * ERROR策略
     * 注意：超过128会报错
     */
    private void errorStrategy() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                /*Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();*/


                for (int i = 0; i < 129; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
                //从128改成129报错了onErrorio.reactivex.exceptions.MissingBackpressureException: create: could not emit value due to lack of requests
                //点击Flowable，进去看源码就会发现最大值是128

            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
//                        s.request(Long.MAX_VALUE);
//                        s.request(2);//按照顺序来的，只接收了发送的1，2两个事件
                        //如果这里设置了s.request(2)
                        //2020-05-06 13:56:24.304 28638-28696/com.example.simplifymvp D/SecondActivity: emit 1
                        //2020-05-06 13:56:24.305 28638-28638/com.example.simplifymvp D/SecondActivity: onNext: 1
                        //2020-05-06 13:56:24.305 28638-28696/com.example.simplifymvp D/SecondActivity: emit 2
                        //2020-05-06 13:56:24.305 28638-28638/com.example.simplifymvp D/SecondActivity: onNext: 2
                        //2020-05-06 13:56:24.305 28638-28696/com.example.simplifymvp D/SecondActivity: emit 3
                        //2020-05-06 13:56:24.305 28638-28696/com.example.simplifymvp D/SecondActivity: emit complete
                    }

                    @Override
                    public void onNext(Integer s) {
                        Log.d(TAG, "onNext: " + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d(TAG, "onError" + t);

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");

                    }
                });
    }


    /***
     * 同步：在同一个线程内
     */
    private void syncSubscribe() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "发送事件 1");
                emitter.onNext(1);
                Log.d(TAG, "发送事件 2");
                emitter.onNext(2);
                Log.d(TAG, "发送事件 3");
                emitter.onNext(3);
                Log.d(TAG, "发送完成");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        s.request(3);
                    }

                    @Override
                    public void onNext(Integer s) {
                        Log.d(TAG, "onNext: " + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d(TAG, "onError" + t);

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");

                    }
                });
    }


    /***
     * LATEST策略
     */
    private void latestStrategy() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {

                for (int i = 0; i < 129 * 3; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                    Thread.sleep(10);
                }
            }
        }, BackpressureStrategy.LATEST).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        subscription = s;
                    }

                    @Override
                    public void onNext(Integer s) {
                        Log.d(TAG, "onNext: " + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d(TAG, "onError" + t);

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");

                    }
                });
    }

    /***
     * DROP策略
     * 只能取出默认的128个缓存池中的数量
     * 我先存128个，当这128个被清空后从新再装进128吧，那么中间这个过程中上游发送的事件，下游就给全部丢掉了。
     * FLowable内部的默认的水缸大小为128, 因此, 它刚开始肯定会把0-127这128个事件保存起来, 然后丢弃掉其余的事件, 当我们request(64)的时候,下游便会处理掉这64个事件，当第二次请求时把水缸里剩余的64个事件清空, 那么上游水缸中又会重新装进新的128个事件。
     */
    private void dropStrategy() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {

                for (int i = 0; i < 500; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.DROP).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        subscription = s;
                    }

                    @Override
                    public void onNext(Integer s) {
                        Log.d(TAG, "onNext: " + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d(TAG, "onError" + t);

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");

                    }
                });
    }

    /***
     * BUFFER策略
     * BUFFER是一个无限大的缓存池，也就是说我们可以往里面存储无限多的事件
     * 但是，如果我们发送无数多的事件，同样要注意内存情况。
     * 注意：只是缓存，下有没有接收到
     */
    private void bufferStrategy() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {

                for (int i = 0; i < 129; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.BUFFER).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {

                    }

                    @Override
                    public void onNext(Integer s) {
                        Log.d(TAG, "onNext: " + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d(TAG, "onError" + t);

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");

                    }
                });
    }

    /***
     * MISSING
     * 会出现错误 onErrorio.reactivex.exceptions.MissingBackpressureException: Queue is full?!
     * 这种策略是不丢弃，不缓存的策略，那么我要你也没什么用啊…………
     * 注意：只是不断的从上游往下游发送事件，超过128会报队列满了。。
     */
    private void missingStrategy() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {

                for (int i = 0; i < 129; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.MISSING).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {

                    }

                    @Override
                    public void onNext(Integer s) {
                        Log.d(TAG, "onNext: " + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d(TAG, "onError" + t);

                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");

                    }
                });
    }
}
